Good point @o-nikolas I was being lazy and didn't want to figure out how to run the test suite. You've pushed me to do the right thing and I think I've added a commit/test that addresses it and fails with the old code while passing with the new. Let me know what you think.

Good point @o-nikolas I was being lazy and didn't want to figure out how to run the test suite. You've pushed me to do the right thing and I think I've added a commit/test that addresses it and fails with the old code while passing with the new. Let me know what you think.

Thanks for adding a test case! A committer will be required to run the workflow for a first time committer (and ultimately merge if everything passes and they are happy with the code). Unfortunately I am not a committer, I'll CC a few who may have time to have a look: @eladkal @potiuk @kaxil

There was a change merged recently to standardize quoting in Airflow, you'll need to rebase this PR and run static checks locally to patch those up.
You can enable pre-commit to easily run static checks on your code (readme).

@o-nikolas I merged in main enabled pre-commit and updated my quoting style so hopefully that's the last of the linting though we'll see for sure after PR checks run. Thanks for the help and heads up that it had changed.

It's fine but I think we need a newsfragment describing this change in the behaviours. This is borderline breaking change/bugfix and while I would lean more on the bugfix side, but people could have relied on it and it should be described in more detail in changelog.

@potiuk because of this bug, to use the ExternalTaskSensor currently you must explicitly set a timeout on the sensor or your DAG will hang forever. To your point on reliance on old behavior, to workaround the bug, folks may have set that timeout to avoid an infinite hang.

In those cases, fixing this bug will cause a change in the exception they receive from AirflowSensorTimeout to the generic AirflowException. If they are relying on catching the AirflowSensorTimeout exception subclass they may have issues though if they catch the base class they'd still be OK.

Does that sound about right? What would you propose we do?

I'm happy to update a changelog if I'm pointed in the right direction?

Also, there are some failing checks on this PR that I don't understand. Specifically, in the Sqlite Py3.7: API Always CLI Core Integration Other Providers WWW check a test fails that I'm pretty sure I don't go anywhere near:

FAILED tests/jobs/test_local_task_job.py::TestLocalTaskJob::test_heartbeat_failed_fast

Any ideas on that front? The logs are long and I didn't see much useful in them while looking through so I wanted to ask before trying to dig deeper as I'm not super familiar with this code-base and the checks on it.

I'm happy to update a changelog if I'm pointed in the right direction?

In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

@eladkal I've added a newsfragment as suggested.

For what it's worth, I think this should break very few if any people because most folks would want their DAGs to fail in the case of an upstream external task failing whether by timeout or because a task in a chain fails. I doubt there are many people who are trying to catch an exception and act on it in this case, but it's at least documented now that you might have to change the exception type in case anyone falls into that bucket.

Can you please rebase/solve conflicts?

@potiuk sorry about that, I had been merging in main instead of rebasing on it, but I've changed that.

Let me know if you need anything else on my end.

@eladkal I noticed that the following test failed for the "Tests / Sqlite Py3.7: API Always CLI Core Integration Other Providers WWW (pull_request)" pre-submit job but I'm pretty sure this test has nothing to do with my code. Does that seem right? Is there anything I should be doing? Or do these tests fail intermittently or something?

Thanks for the help!

=================================== FAILURES ===================================
_________________ TestLocalTaskJob.test_heartbeat_failed_fast __________________

self = <tests.jobs.test_local_task_job.TestLocalTaskJob object at 0x7f1378e24390>

    def test_heartbeat_failed_fast(self):
        """
        Test that task heartbeat will sleep when it fails fast
        """
        self.mock_base_job_sleep.side_effect = time.sleep
        dag_id = "test_heartbeat_failed_fast"
        task_id = "test_heartbeat_failed_fast_op"
        with create_session() as session:
    
            dag_id = "test_heartbeat_failed_fast"
            task_id = "test_heartbeat_failed_fast_op"
            dag = self.dagbag.get_dag(dag_id)
            task = dag.get_task(task_id)
    
            dr = dag.create_dagrun(
                run_id="test_heartbeat_failed_fast_run",
                state=State.RUNNING,
                execution_date=DEFAULT_DATE,
                start_date=DEFAULT_DATE,
                session=session,
            )
    
            ti = dr.task_instances[0]
            ti.refresh_from_task(task)
            ti.state = State.QUEUED
            ti.hostname = get_hostname()
            ti.pid = 1
            session.commit()
    
            job = LocalTaskJob(task_instance=ti, executor=MockExecutor(do_update=False))
            job.heartrate = 2
            heartbeat_records = []
            job.heartbeat_callback = lambda session: heartbeat_records.append(job.latest_heartbeat)
            job._execute()
            assert len(heartbeat_records) > 2
            for i in range(1, len(heartbeat_records)):
                time1 = heartbeat_records[i - 1]
                time2 = heartbeat_records[i]
                # Assert that difference small enough
                delta = (time2 - time1).total_seconds()
>               assert abs(delta - job.heartrate) < 0.8
E               assert 1.0317020000000001 < 0.8
E                +  where 1.0317020000000001 = abs((3.031702 - 2))
E                +    where 2 = <airflow.jobs.local_task_job.LocalTaskJob object at 0x7f1378e35cd0>.heartrate

tests/jobs/test_local_task_job.py:312: AssertionError

I reran the test. lets see what happens

@eladkal that seems to have done the trick, thanks for the help!

Anything else you need on my end to get this merged?

Are we getting this into 2.4.3? @eladkal @potiuk @uranusjr

Awesome work, congrats on your first merged pull request!